package paxos

//
// Paxos library, to be included in an application.
// Multiple applications will run, each including
// a Paxos peer.
//
// Manages a sequence of agreed-on values.
// The set of peers is fixed.
// Copes with network failures (partition, msg loss, &c).
// Does not store anything persistently, so cannot handle crash+restart.
//
// The application interface:
//
// px = paxos.Make(peers []string, me string)
// px.Start(seq int, v interface{}) -- start agreement on new instance
// px.Status(seq int) (decided bool, v interface{}) -- get info about an instance
// px.Done(seq int) -- ok to forget all instances <= seq
// px.Max() int -- highest instance seq known, or -1
// px.Min() int -- instances before this seq have been forgotten
//

import "net"
import "net/rpc"
import "log"
import "os"
import "syscall"
import "sync"
import "fmt"
import "math/rand"
import "time"
import "atomutil"
import "strconv"
import "path"
import "path/filepath"
import "strings"
import "regexp"
import "util"

type PaxosState struct {
	Np int
	Na int
	Va interface{}

	Decided  bool
	DecidedV interface{}

	InsId int // instance id
}

type Paxos struct {
	mu         sync.Mutex
	l          net.Listener
	dead       bool
	unreliable bool
	peers      []string
	me         int // index into peers[]

	// Your data here.
	acLocks map[int]*sync.Mutex // mutex for acceptor
	ppLocks map[int]*sync.Mutex // mutex for proposer

	mapStates map[int]PaxosState

	doneValues []int
	maxValue   int
	outputPath string

	logger *util.Logger

	// for tcp
	deaf bool   // to simulate tcp deaf
	tag  string // for persistency
}

type PrepareArgs struct {
	Insnum int
	N      int

	From      int
	DoneValue int
}

type PrepareReply struct {
	Msg string
	Np  int
	Na  int
	Va  interface{}
}

type AcceptArgs struct {
	Insnum int
	N      int
	V      interface{}

	From      int
	DoneValue int
}

type AcceptReply struct {
	Msg string
	Va  interface{}
}

type DecidedArgs struct {
	Insnum int
	V      interface{}

	From      int
	DoneValue int
}

type DecidedReply struct {
}

const (
	PREPARE_OK     = "prepare_ok"
	PREPARE_REJECT = "prepare_reject"
	PREPARE_ERROR  = "prepare_error"
	DECIDED        = "decided"
	ACCEPT_OK      = "accept_ok"
	ACCEPT_REJECT  = "accept_reject"
	ACCEPT_ERROR   = "accept_error"

	SleepInterval = time.Millisecond * 10
	DEBUG         = false
	//DEBUG      = true
	PERSISTENT = true
)

//
// call() sends an RPC to the rpcname handler on server srv
// with arguments args, waits for the reply, and leaves the
// reply in reply. the reply argument should be a pointer
// to a reply structure.
//
// the return value is true if the server responded, and false
// if call() was not able to contact the server. in particular,
// the replys contents are only valid if call() returned true.
//
// you should assume that call() will time out and return an
// error after a while if it does not get a reply from the server.
//
// please use call() to send all RPCs, in client.go and server.go.
// please do not change this function.
//
func call(srv string, name string, args interface{}, reply interface{}) bool {
	// a little hacky: instead of passing local var/setting global var
	var errx error
	var c *rpc.Client
	if strings.Contains(srv, ":") {
		c, errx = rpc.Dial("tcp", srv)
	} else {
		c, errx = rpc.Dial("unix", srv)
	}
	if errx != nil {
		return false
	}
	defer c.Close()

	err := c.Call(name, args, reply)
	if err == nil {
		return true
	}
	return false
}

func (px *Paxos) GetAcceptorLock(seq int) *sync.Mutex {

	px.mu.Lock()
	defer px.mu.Unlock()

	mu, exist := px.acLocks[seq]
	if !exist {
		mu = &sync.Mutex{}
		px.acLocks[seq] = mu
	}

	return mu
}

func (px *Paxos) GetProposerLock(seq int) *sync.Mutex {

	px.mu.Lock()
	defer px.mu.Unlock()

	mu, exist := px.ppLocks[seq]
	if !exist {
		mu = &sync.Mutex{}
		px.ppLocks[seq] = mu
	}

	return mu

}

//
// create a new paxos instance state if it doesn't exist
//
func (px *Paxos) checkOrCreateState(seq int) (PaxosState, error) {

	px.mu.Lock()
	defer px.mu.Unlock()

	if px.maxValue == -2 {
		px.InitMaxValue()
	}
	if seq > px.maxValue {
		px.maxValue = seq
	}

	cState, bExist := px.mapStates[seq]
	if bExist == false {
		if PERSISTENT {
			path := px.outputPath + ".paxos." + strconv.Itoa(seq)
			exist, err := atomutil.IsDataExist(path)
			if err != nil {
				return PaxosState{}, err
			}

			if exist == false {
				cState = PaxosState{}
				cState.Decided = false
				cState.DecidedV = nil
				cState.Np = -1
				cState.Na = -1
				cState.Va = nil
				err = atomutil.WriteData(path, &cState)
				if err != nil {
					return PaxosState{}, err
				}
				px.mapStates[seq] = cState
			} else {
				err = atomutil.ReadData(path, &cState)
				if err != nil {
					return PaxosState{}, err
				}
				px.mapStates[seq] = cState
			}
		} else {
			cState = PaxosState{}
			cState.Decided = false
			cState.DecidedV = nil
			cState.Np = -1
			cState.Na = -1
			cState.Va = nil
			px.mapStates[seq] = cState
		}
	}
	//if(bExist && px.me==0){fmt.Println("cc:",cState)}
	return cState, nil

}

//
// set paxos instance state
//
func (px *Paxos) setState(seq int, state PaxosState) error {

	px.mu.Lock()
	defer px.mu.Unlock()

	if PERSISTENT {
		path := px.outputPath + ".paxos." + strconv.Itoa(seq)
		err := atomutil.WriteData(path, &state)
		if err != nil {
			return err
		}
	}
	px.mapStates[seq] = state

	return nil

}

//
//
//
func (px *Paxos) getDoneValue() int {

	px.mu.Lock()
	defer px.mu.Unlock()

	return px.doneValues[px.me]
}

//
//
//
func (px *Paxos) updateDoneValue(from int, value int) {

	px.mu.Lock()
	defer px.mu.Unlock()

	if px.doneValues[from] < value {
		px.doneValues[from] = value
	}
}

//
// acceptor's prepare handler
//
func (px *Paxos) Prepare(args *PrepareArgs, reply *PrepareReply) error {

	iSeq := args.Insnum
	iN := args.N

	acMu := px.GetAcceptorLock(iSeq)
	acMu.Lock()
	defer acMu.Unlock()

	cState, err := px.checkOrCreateState(args.Insnum)
	if err != nil {
		fmt.Printf("Paxos Read/Write Failed: %v\n", err)
		reply.Msg = PREPARE_ERROR
		return nil
	}
	px.updateDoneValue(args.From, args.DoneValue)

	if cState.Decided {
		//fmt.Printf("already deicided.\n")
		reply.Msg = DECIDED
		reply.Va = cState.DecidedV
		return nil
	}

	if iN > cState.Np {
		if DEBUG {
			fmt.Printf("Ins %d Ac %d prepare_ok Np=%v<%v Na=%v Va=%v\n", iSeq,
				px.me, cState.Np, iN, cState.Na, cState.Va)
		}
		cState.Np = iN
		err = px.setState(iSeq, cState)
		if err != nil {
			fmt.Printf("Paxos Read/Write Failed: %v\n", err)
			reply.Msg = PREPARE_ERROR
			return nil
		}
		reply.Msg = PREPARE_OK
		reply.Na = cState.Na
		reply.Va = cState.Va
	} else {
		if DEBUG {
			fmt.Printf("Ins %d Ac %d prepare_reject Np=%v>%v Na=%v Va=%v\n", iSeq,
				px.me, cState.Np, iN, cState.Na, cState.Va)
		}
		reply.Msg = PREPARE_REJECT
		reply.Np = cState.Np
	}

	return nil
}

//
// acceptor's accept handler
//
func (px *Paxos) Accept(args *AcceptArgs, reply *AcceptReply) error {

	iSeq := args.Insnum
	iN := args.N
	cV := args.V

	acMu := px.GetAcceptorLock(iSeq)
	acMu.Lock()
	defer acMu.Unlock()

	cState, err := px.checkOrCreateState(args.Insnum)
	if err != nil {
		fmt.Printf("Paxos Read/Write Failed: %v\n", err)
		reply.Msg = ACCEPT_ERROR
		return nil
	}
	px.updateDoneValue(args.From, args.DoneValue)

	if cState.Decided {
		//fmt.Printf("already deicided.\n")
		reply.Msg = DECIDED
		reply.Va = cState.DecidedV
		return nil
	}

	if iN >= cState.Np {
		if DEBUG {
			fmt.Printf("Ins %d Ac %d accept_ok Np=%v<%v Na=%v Va=%v V=%v\n", iSeq,
				px.me, cState.Np, iN, cState.Na, cState.Va, cV)
		}
		cState.Np = iN
		cState.Na = iN
		cState.Va = cV
		err = px.setState(iSeq, cState)
		if err != nil {
			fmt.Printf("Paxos Read/Write Failed: %v\n", err)
			reply.Msg = ACCEPT_ERROR
			return nil
		}
		reply.Msg = ACCEPT_OK
	} else {
		if DEBUG {
			fmt.Printf("Ins %d Ac %d accept_reject Np=%v>%v Na=%v Va=%v V=%v\n", iSeq,
				px.me, cState.Np, iN, cState.Na, cState.Va, cV)
		}
		reply.Msg = ACCEPT_REJECT
	}

	return nil
}

//
// acceptor's decided handler
//
func (px *Paxos) Decided(args *DecidedArgs, reply *DecidedReply) error {

	iSeq := args.Insnum
	cV := args.V

	acMu := px.GetAcceptorLock(iSeq)
	acMu.Lock()
	defer acMu.Unlock()

	cState, err := px.checkOrCreateState(args.Insnum)
	if err != nil {
		fmt.Printf("Paxos Read/Write Failed: %v\n", err)
		return nil
	}
	px.updateDoneValue(args.From, args.DoneValue)

	if cState.Decided {
		return nil
	}
	cState.Decided = true
	cState.DecidedV = cV
	err = px.setState(iSeq, cState)
	if err != nil {
		fmt.Printf("Paxos Read/Write Failed: %v\n", err)
		return nil
	}

	return nil
}

//
// proposer function
//
func (px *Paxos) proposer(seq int, v interface{}) {

	var cState PaxosState
	var err error
	for {
		cState, err = px.checkOrCreateState(seq)
		if err == nil {
			break
		}
		fmt.Printf("Paxos Failed to Read State in proposer function: %v\n", err)
		time.Sleep(SleepInterval)
	}

	ppMu := px.GetProposerLock(seq)
	ppMu.Lock()
	defer ppMu.Unlock()

	if cState.Decided {
		return
	}

	bDecided := false
	cV := v
	iMaxN := -1
	iN := px.me
	iSize := len(px.peers)
	iNnext := px.me + iSize
	iMajority := (iSize + 2) / 2
	for {
		bDecided = false
		iMaxN = -1
		iN = iNnext
		iNnext += iSize // increase uid

		if DEBUG {
			fmt.Printf("Ins %d Pp %d prepare N=%d ...\n", seq, px.me, iN)
		}

		// Prepare Phrase
		iReceived := 0
		for i := 0; i < iSize; i++ {
			if iReceived+(iSize-i) < iMajority {
				break
			}

			prepareArgs := &PrepareArgs{}
			prepareArgs.Insnum = seq
			prepareArgs.N = iN
			prepareArgs.From = px.me
			prepareArgs.DoneValue = px.getDoneValue()

			var prepareReply PrepareReply
			var ok bool

			if i == px.me {
				ok = true
				px.Prepare(prepareArgs, &prepareReply)
			} else {
				ok = call(px.peers[i], "Paxos.Prepare", prepareArgs, &prepareReply)
			}

			if ok == true {
				if prepareReply.Msg == DECIDED {
					bDecided = true
					cV = prepareReply.Va
					break
				}
				if prepareReply.Msg == PREPARE_OK {
					iReceived += 1
					if iMaxN < prepareReply.Na {
						iMaxN = prepareReply.Na
						cV = prepareReply.Va
					}
				}
				if prepareReply.Msg == PREPARE_REJECT {

					// increase uid so it is greater than previous promise next time
					iNnext = (prepareReply.Np/iSize+1)*iSize + px.me

					//time.Sleep(time.Duration(rand.Int() % 20 + 1) * SleepInterval)
					time.Sleep(time.Duration((px.me-1)%10+1) * SleepInterval)

					continue
				}
			}
		}

		if bDecided == false {

			if DEBUG {
				fmt.Printf("Ins %d Pp %d prepare N=%d OK=%d\n", seq, px.me, iN, iReceived)
			}

			if iReceived < iMajority {
				//time.Sleep(time.Duration(rand.Int() % 20 + 1) * SleepInterval)
				time.Sleep(time.Duration((px.me-1)%10+1) * SleepInterval)
				continue
			}

			if DEBUG {
				fmt.Printf("Ins %d Pp %d accept N=%d V=%v ...\n", seq, px.me, iN, cV)
			}

			// Accept Phrase
			iReceived = 0
			for i := 0; i < iSize; i++ {
				if iReceived+(iSize-i) < iMajority {
					break
				}

				acceptArgs := &AcceptArgs{}
				acceptArgs.Insnum = seq
				acceptArgs.N = iN
				acceptArgs.V = cV
				acceptArgs.From = px.me
				acceptArgs.DoneValue = px.getDoneValue()

				var acceptReply AcceptReply
				var ok bool

				if i == px.me {
					ok = true
					px.Accept(acceptArgs, &acceptReply)
				} else {
					ok = call(px.peers[i], "Paxos.Accept", acceptArgs, &acceptReply)
				}

				if ok == true {
					if acceptReply.Msg == DECIDED {
						bDecided = true
						cV = acceptReply.Va
						break
					}
					if acceptReply.Msg == ACCEPT_OK {
						iReceived += 1
					}
				}
			}

			if DEBUG {
				fmt.Printf("Ins %d Pp %d accept N=%d V=%v OK=%v\n", seq, px.me, iN, cV, iReceived)
			}

		}

		if bDecided || iReceived >= iMajority {
			if DEBUG {
				fmt.Printf("Ins %d Pp %d decided V=%v\n", seq, px.me, cV)
			}
			for i := 0; i < iSize; i++ {
				decidedArgs := &DecidedArgs{}
				decidedArgs.Insnum = seq
				decidedArgs.V = cV
				decidedArgs.From = px.me
				decidedArgs.DoneValue = px.getDoneValue()
				var decidedReply DecidedReply
				var ok bool

				if i == px.me {
					ok = true
					px.Decided(decidedArgs, &decidedReply)
				} else {
					ok = call(px.peers[i], "Paxos.Decided", decidedArgs, &decidedReply)
				}

				if ok == true {
				}
			}

			break
		}

		//time.Sleep(time.Duration(rand.Int() % 20 + 1) * SleepInterval)
		time.Sleep(time.Duration((px.me-1)%10+1) * SleepInterval)
	}

}

//
// the application wants paxos to start agreement on
// instance seq, with proposed value v.
// Start() returns right away; the application will
// call Status() to find out if/when agreement
// is reached.
//
func (px *Paxos) Start(seq int, v interface{}) {

	if seq < px.Min() {
		return
	}

	px.mu.Lock()
	defer px.mu.Unlock()

	//_, bExist := px.mapStates[seq]
	//if bExist == true {
	//    fmt.Printf("Error: instance %d already exists for px %d!\n", seq, px.me)
	//    return
	//}

	if DEBUG {
		fmt.Printf("Server %v start ins %v v=%v\n", px.me, seq, v)
	}
	// start a proposer for this instance
	go px.proposer(seq, v)
}

//
// the application on this machine is done with
// all instances <= seq.
//
// see the comments for Min() for more explanation.
//
func (px *Paxos) Done(seq int) {

	px.mu.Lock()
	defer px.mu.Unlock()

	if px.doneValues[px.me] < seq+1 {
		px.doneValues[px.me] = seq + 1
	}

	px.logger.Logger.Printf("Done: %v\n", px.doneValues)
}

func trimLeadingDot(filename string) string {
	if strings.HasPrefix(filename, "./") {
		filename = filename[2:]
	}
	return filename
}
func (px *Paxos) ExtractPaxosInstanceNumber(file string) (int, error) {
	s := file[len(px.outputPath+".paxos."):]
	if pos := strings.Index(s, "."); pos != -1 {
		s = s[:pos]
	}
	n, err2 := strconv.Atoi(s)
	return n, err2
}

//
// the application wants to know the
// highest instance sequence known to
// this peer.
//
func (px *Paxos) Max() int {

	px.mu.Lock()
	defer px.mu.Unlock()

	//return px.maxValue

	if px.maxValue == -2 {
		px.InitMaxValue()
	}

	return px.maxValue
}

func (px *Paxos) InitMaxValue() {

	if PERSISTENT == false {
		px.maxValue = -1
		return
	}

	var lstFiles []string
	var err error
	for {
		lstFiles, err = filepath.Glob(px.outputPath + ".paxos.*")
		if err == nil {
			break
		}
		fmt.Printf("Paxos Failed to Init MaxValue: %v\n", err)
		time.Sleep(SleepInterval)
	}

	if lstFiles == nil || len(lstFiles) == 0 {
		px.maxValue = -1
		return
	}
	maxN := -1
	for _, file := range lstFiles {
		n, err2 := px.ExtractPaxosInstanceNumber(file)
		if err2 != nil {
			fmt.Printf("Warning: failed to parse ins#: %v\n", file)
		} else {
			if n > maxN {
				maxN = n
			}
		}
	}

	px.maxValue = maxN

}

//
// Min() should return one more than the minimum among z_i,
// where z_i is the highest number ever passed
// to Done() on peer i. A peers z_i is -1 if it has
// never called Done().
//
// Paxos is required to have forgotten all information
// about any instances it knows that are < Min().
// The point is to free up memory in long-running
// Paxos-based servers.
//
// It is illegal to call Done(i) on a peer and
// then call Start(j) on that peer for any j <= i.
//
// Paxos peers need to exchange their highest Done()
// arguments in order to implement Min(). These
// exchanges can be piggybacked on ordinary Paxos
// agreement protocol messages, so it is OK if one
// peers Min does not reflect another Peers Done()
// until after the next instance is agreed to.
//
// The fact that Min() is defined as a minimum over
// *all* Paxos peers means that Min() cannot increase until
// all peers have been heard from. So if a peer is dead
// or unreachable, other peers Min()s will not increase
// even if all reachable peers call Done. The reason for
// this is that when the unreachable peer comes back to
// life, it will need to catch up on instances that it
// missed -- the other peers therefor cannot forget these
// instances.
// 
func (px *Paxos) Min() int {

	px.mu.Lock()
	defer px.mu.Unlock()

	min := px.doneValues[0]
	for i := 1; i < len(px.peers); i++ {
		if px.doneValues[i] < min {
			min = px.doneValues[i]
		}
	}

	px.logger.Logger.Printf("%v: min=%v\n", px.outputPath, min)

	var insId2Remove []int
	for insId := range px.mapStates {
		if insId < min {
			insId2Remove = append(insId2Remove, insId)
		}
	}
	for _, insId := range insId2Remove {
		delete(px.mapStates, insId)
		if PERSISTENT {
			os.Remove(px.outputPath + ".paxos." + strconv.Itoa(insId))
		}
	}

	return min
}

//
// the application wants to know whether this
// peer thinks an instance has been decided,
// and if so what the agreed value is. Status()
// should just inspect the local peers state;
// it should not contact other Paxos peers.
//
func (px *Paxos) Status(seq int) (bool, interface{}) {

	if seq < px.Min() {
		return false, nil
	}

	var cState PaxosState
	var err error
	for {
		cState, err = px.checkOrCreateState(seq)
		if err == nil {
			break
		}
		fmt.Printf("Paxos Failed to Read State in Status function: %v\n", err)
		time.Sleep(SleepInterval)
	}

	if cState.Decided == true {
		return true, cState.DecidedV
	}

	return false, nil
}

//
// tell the peer to shut itself down.
// for testing.
// please do not change this function.
//

func (px *Paxos) Kill() {
	px.dead = true
	if px.l != nil {
		px.l.Close()
		//fmt.Println("close: ", px.me, px.peers[px.me])
	}
}
func (px *Paxos) CleanDir() {
	os.RemoveAll(path.Dir(px.outputPath))
	fmt.Println("clean:", path.Dir(px.outputPath))
}

func (px *Paxos) KillEar() {
	px.deaf = true
}
func (px *Paxos) RecoverEar() {
	px.deaf = false
}
func (px *Paxos) SetTag(tag string) {
	px.tag = tag
	//update tag name
	os.RemoveAll(path.Dir(px.outputPath))
    //fmt.Println("rm: ",path.Dir(px.outputPath))
	px.outputPath = px.CreateOutputPath(px.peers[px.me])
}
func (px *Paxos) CreateOutputPath(peer string) string {
	pname := peer
	if strings.Contains(peer, ":") {
		// need to create a tmp directory
		// need random number to differentiate
		pname = "/var/tmp/824-"
		pname += strconv.Itoa(os.Getuid()) + "/"
		os.Mkdir(pname, 0777)
		if px.tag == "" {
			// needed to differentiate between tests
			pname += peer + time.Now().String()
			pname = strings.Replace(pname, ".", "_", -1)
			pname = strings.Replace(pname, ":", "_", -1)
			pname = strings.Replace(pname, " ", "_", -1)
		} else {
			pname += "px-" + px.tag + px.peers[px.me]
		}
		//pname += peer+strconv.Itoa(int(time.Now().Unix()))
		//fmt.Println("ap:", pname, peer)
	}
	dir := path.Dir(pname)
	base := path.Base(pname)
	os.Mkdir(dir+"/"+base+".workdir", 0777)
	path := dir + "/" + base + ".workdir/" + base
	path = trimLeadingDot(path)
	return path
}

func IsIP(addr string) bool {
	matched, err := regexp.MatchString(`[^:]*:\d+`, addr)
	return matched && err == nil
}

//
// the application wants to create a paxos peer.
// the ports of all the paxos peers (including this one)
// are in peers[]. this servers port is peers[me].
//
func Make(peers []string, me int, rpcs *rpc.Server) *Paxos {
	px := &Paxos{}
	px.peers = peers
	px.deaf = false
	px.me = me

	// Your initialization code here.
	px.mapStates = map[int]PaxosState{}
	px.acLocks = map[int]*sync.Mutex{}
	px.ppLocks = map[int]*sync.Mutex{}
	px.doneValues = make([]int, len(peers))
	//px.maxValue = -1
	px.maxValue = -2
	px.tag = ""
	for i := 0; i < len(peers); i++ {
		px.doneValues[i] = 0
	}
	px.outputPath = px.CreateOutputPath(px.peers[me])
	//fmt.Println("so:",px.outputPath)

	px.logger = util.NewLogger(px.peers[me] + ".px", false)

	if rpcs != nil {
		// caller will create socket &c
		rpcs.Register(px)
	} else {
		rpcs = rpc.NewServer()
		rpcs.Register(px)

		// prepare to receive connections from clients.
		var l net.Listener
		var e error
		port := peers[me]
		if IsIP(port) {
			fmt.Printf("Listening to TCP %v\n", port)
			l, e = net.Listen("tcp", port)
		} else {
			fmt.Printf("Listening to File %v\n", port)
			os.Remove(port)
			l, e = net.Listen("unix", port)
		}
		if e != nil {
			log.Fatal("listen error: ", e)
		}
		px.l = l

		// please do not change any of the following code,
		// or do anything to subvert it.

		// create a thread to accept RPC connections
		go func() {
			for px.dead == false {
				conn, err := px.l.Accept()
				if err == nil && px.dead == false {
					if px.deaf || (px.unreliable && (rand.Int63()%1000) < 100) {
						// discard the request.
						conn.Close()
					} else if px.unreliable && (rand.Int63()%1000) < 200 {
						// process the request but force discard of reply.
						var f *os.File
						if !IsIP(port) {
							c1 := conn.(*net.UnixConn)
							f, _ = c1.File()
						}
						err := syscall.Shutdown(int(f.Fd()), syscall.SHUT_WR)
						if err != nil {
							fmt.Printf("shutdown: %v\n", err)
						}
						go rpcs.ServeConn(conn)
					} else {
						go rpcs.ServeConn(conn)
					}
				} else if err == nil {
					conn.Close()
				}
				if err != nil && px.dead == false {
					fmt.Printf("Paxos(%v) accept: %v\n", me, err.Error())
					px.Kill()
				}
			}
		}()
	}

	return px
}
